package com.zhss.dfs.server;

import javafx.concurrent.Worker;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.CacheRequest;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @program: filesystem
 * @description:
 * @author: Liu Weidong
 * @create: 2024-05-14 21:58
 **/
public class NioFileServer extends Thread {

    public static final Integer READ_TYPE = 1;
    public static final Integer SEND_TYPE = 2;

    private Selector selector;

    /**
     * 缓存上传拆包问题
     */
    private Map<String, CacheRequest> cacheRequest = new ConcurrentHashMap<>();

    /**
     * 缓存请求类型
     */
    private Map<String, ByteBuffer> cacheRequestType = new ConcurrentHashMap<>();

    /**
     * 缓存文件名长度信息
     */
    private Map<String, ByteBuffer> cacheFileNameLength = new ConcurrentHashMap<>();

    /**
     * 缓存文件名
     */
    private Map<String, ByteBuffer> cacheFileName = new ConcurrentHashMap<>();

    /**
     * 缓存文件长度
     */
    private Map<String, ByteBuffer> cacheFileLength = new ConcurrentHashMap<>();

    /**
     *  缓存文件内容
     */
    private Map<String, ByteBuffer> cacheFileContent = new ConcurrentHashMap<>();

    private NameNodeServiceActor nameNodeServiceActor;


    public NioFileServer(NameNodeServiceActor nameNodeServiceActor) {
        ServerSocketChannel serverSocketChannel = null;
        this.nameNodeServiceActor = nameNodeServiceActor;
        try {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.bind(new InetSocketAddress(DataNodeConfig.DATANODE_HOSTNAME, DataNodeConfig.DATANODE_PORT));
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        try {
            System.out.println("dataNode-fileServer已经启动监听端口号" + DataNodeConfig.DATANODE_PORT + "...........................");
            while (true) {
                selector.select();
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();
                    handleEvents(key);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    private void handleEvents(SelectionKey key) throws IOException {
        SocketChannel socketChannel = null;
        try {
            if (key.isAcceptable()) {
                ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                socketChannel = serverSocketChannel.accept();
                socketChannel.configureBlocking(false);
                if (socketChannel != null) {
                    socketChannel.configureBlocking(false);
                    socketChannel.register(selector, SelectionKey.OP_READ);
                    System.out.println("接收到连接请求，注册读事件。。。。。。。。。。。");
                }
            } else if (key.isReadable()) {
                socketChannel = (SocketChannel) key.channel();
                handleRequest(socketChannel, key);
            }
        } catch (Exception e) {
            e.printStackTrace();
            if (socketChannel != null) {
                socketChannel.close();
            }
        }
    }


    private void handleRequest(SocketChannel socketChannel, SelectionKey key) throws IOException {
        int requestType = getRequestType(socketChannel);
        if (requestType == SEND_TYPE) {
            handleSendFileRequest(socketChannel, key);
        } else if (requestType == READ_TYPE) {
            sendFileToClient(socketChannel, key);
        } else if (requestType == -1) {
            socketChannel.close();
        }
    }


    private void sendFileToClient(SocketChannel socketChannel,SelectionKey key) throws IOException {
        String client = socketChannel.getRemoteAddress().toString();
        // 获取文件名的相对路径
        String relativeFileName = getRelativeFileName(socketChannel);
        if (relativeFileName == null) {
            socketChannel.close();
            return;
        }
        // 根据filePath获取本地文件
        // 获取文件名的绝对路径
        String absolutePath = DataNodeConfig.FILE_DATA + relativeFileName;
        System.out.println("文件的绝对路径为：" + absolutePath);
        FileInputStream inputStream = new FileInputStream(absolutePath);
        FileChannel inputStreamChannel = inputStream.getChannel();
        ByteBuffer buffer = ByteBuffer.allocate(8 + Long.valueOf(inputStreamChannel.size()).intValue());
        Long fileLength = inputStreamChannel.size();
        buffer.putLong(fileLength);
        int hasReadImageLength = inputStreamChannel.read(buffer);
        System.out.println("从本次磁盘文件中读取了" + hasReadImageLength + " bytes的数据");

        buffer.rewind();
        int sent = socketChannel.write(buffer);
        System.out.println("将" + sent + " bytes的数据发送给了客户端.....");

        inputStreamChannel.close();
        inputStream.close();

        // 判断一下，如果已经读取完毕，就返回一个成功给客户端
        if(hasReadImageLength == fileLength) {
            System.out.println("文件发送完毕，给客户端: " + client);
            cacheRequest.remove(client);
            key.interestOps(key.interestOps() & ~SelectionKey.OP_READ);
        }
    }

    /**
     * 处理上传文件的请求
     *
     * @param socketChannel
     * @throws IOException
     */
    private void handleSendFileRequest(SocketChannel socketChannel,SelectionKey key) throws IOException {
        String client = socketChannel.getRemoteAddress().toString();
        String relativeFileName = null;
        if (getCachedRequest(client).fileName != null) {
            relativeFileName = getCachedRequest(client).fileName;
        } else {
            // 获取文件名的相对路径
            relativeFileName = getRelativeFileName(socketChannel);
        }
        if (relativeFileName == null) {
            socketChannel.close();
            return;
        }
        // 获取文件名的绝对路径
        String absoluteFileName = getAbsoluteFileName(relativeFileName);
        Long fileLength = cacheRequest.get(client).fileLength != null ? cacheRequest.get(client).fileLength : getFileLength(socketChannel);
        if (fileLength == null) {
            return;
        }
        long hasReadFileLength = cacheRequest.get(client).hasReadFileLength != null ? cacheRequest.get(client).hasReadFileLength : 0L;
        // 向文件目录写入文件
        FileOutputStream outputStream = null;
        FileChannel fileChannel = null;
        try {
            outputStream = new FileOutputStream(absoluteFileName);
            fileChannel = outputStream.getChannel();
            fileChannel.position(fileChannel.size());
            ByteBuffer fileContentByteBuffer = null;
            if (cacheFileContent.containsKey(client)) {
                fileContentByteBuffer = cacheFileContent.get(client);
            } else {
                fileContentByteBuffer = ByteBuffer.allocate(fileLength.intValue());
            }
            hasReadFileLength += socketChannel.read(fileContentByteBuffer);
            if (!fileContentByteBuffer.hasRemaining()) {
                // 文件内容已经全部读取完毕
                fileContentByteBuffer.rewind();
                fileChannel.write(fileContentByteBuffer);
                cacheFileContent.remove(client);

                // 返回正确信息
                ByteBuffer outBuffer = ByteBuffer.wrap("SUCCESS".getBytes());
                socketChannel.write(outBuffer);
                // 移除本次文件上传的缓存
                cacheRequest.remove(client);
                System.out.println("文件读取完毕.......");
                // 向nameNode上报副本信息
                nameNodeServiceActor.InformReceivedReplicateInfo(absoluteFileName, DataNodeConfig.DATANODE_HOSTNAME, DataNodeConfig.DATANODE_IP,fileLength);
                System.out.println("增量上报收到文件副本给NameNode节点.....");
                key.interestOps(key.interestOps() & ~SelectionKey.OP_READ);
            } else {
                cacheFileContent.put(client, fileContentByteBuffer);
                getCachedRequest(client).hasReadFileLength = hasReadFileLength;
                return;
            }
        } finally {
            fileChannel.close();
            outputStream.close();
        }
    }

    /**
     * 获取缓存的请求
     * @param client
     * @return
     */
    private CacheRequest getCachedRequest(String client) {
        CacheRequest cachedRequest = cacheRequest.get(client);
        if(cachedRequest == null) {
            cachedRequest = new CacheRequest();
            cacheRequest.put(client, cachedRequest);
        }
        return cachedRequest;
    }

    private int getRequestType(SocketChannel socketChannel) throws IOException {
        String remoteAddr = socketChannel.getRemoteAddress().toString();
        if(getCachedRequest(remoteAddr).requestType != null) {
            return getCachedRequest(remoteAddr).requestType;
        }
        ByteBuffer byteBuffer = null;
        if (cacheRequestType.containsKey(remoteAddr)) {
            byteBuffer = cacheRequestType.get(remoteAddr);
        } else {
            byteBuffer = ByteBuffer.allocate(4);
        }
        socketChannel.read(byteBuffer);
        if (!byteBuffer.hasRemaining()) {
            byteBuffer.rewind();
            int requestType = byteBuffer.getInt();
            cacheRequestType.remove(remoteAddr);
            CacheRequest cacheRequest = this.cacheRequest.get(remoteAddr);
            cacheRequest.requestType = requestType;
            return requestType;
        } else {
            // 没有读完拆包了
            cacheRequestType.put(remoteAddr, byteBuffer);
        }
        return -1;
    }

    private String getAbsoluteFileName(String filePath) {
        String parentDir = DataNodeConfig.FILE_DATA +
                filePath.substring(0, filePath.lastIndexOf("/"));
        File file = new File(parentDir);
        if (!file.exists()) {
            file.mkdirs();
        }
        return DataNodeConfig.FILE_DATA + filePath;
    }

    private long getFileLength(SocketChannel socketChannel) throws IOException {
        Long fileLength = null;
        ByteBuffer fileLengthBuffer = null;
        String remoteAddr = socketChannel.getRemoteAddress().toString();
        if (cacheFileLength.containsKey(remoteAddr)) {
            fileLengthBuffer = cacheFileLength.get(remoteAddr);
        } else {
            fileLengthBuffer = ByteBuffer.allocate(8);
        }

        socketChannel.read(fileLengthBuffer);
        if (!fileLengthBuffer.hasRemaining()) {
            fileLengthBuffer.rewind();
            fileLength = fileLengthBuffer.getLong();
            cacheRequest.get(remoteAddr).fileLength = fileLength;
            cacheFileLength.remove(remoteAddr);
            System.out.println("文件程度为： " + fileLength);
        } else {
            cacheFileLength.put(remoteAddr, fileLengthBuffer);
        }
        return fileLength;
    }

    private String getRelativeFileName(SocketChannel channel) throws IOException {
        Integer fileNameLength = null;
        String remoteAddr = channel.getRemoteAddress().toString();
        ByteBuffer buffer = null;
        // 读取文件名长度
        if (cacheFileNameLength.containsKey(remoteAddr)) {
            buffer = cacheFileNameLength.get(remoteAddr);
        } else {
            buffer = ByteBuffer.allocate(4);
        }
        channel.read(buffer);
        if (!buffer.hasRemaining()) {
            buffer.rewind();
            fileNameLength = buffer.getInt();
            cacheFileNameLength.remove(remoteAddr);
            System.out.println("文件名长度为：" + fileNameLength);
        } else {
            cacheFileNameLength.put(remoteAddr, buffer);
            return null;
        }

        // 读取文件名
        java.lang.String fileName = null;
        ByteBuffer fileNameBuffer = null;
        if (cacheFileName.containsKey(remoteAddr)) {
            fileNameBuffer = cacheFileName.get(remoteAddr);
        } else {
            fileNameBuffer = ByteBuffer.allocate(fileNameLength);
        }
        channel.read(fileNameBuffer);
        if (!fileNameBuffer.hasRemaining()) {
            fileNameBuffer.rewind();
            fileName = new java.lang.String(fileNameBuffer.array());
            cacheRequest.get(remoteAddr).fileName = fileName;
            cacheFileName.remove(remoteAddr);
        } else {
            cacheFileName.put(remoteAddr, fileNameBuffer);
            return fileName;
        }
        return fileName;
    }

    class CacheRequest {
        java.lang.String fileName;
        Integer requestType;
        Long fileLength;
        Long hasReadFileLength;
    }
}
